package org.example.partition_markdone;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonPartitionMarkDoneTest {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sql1 = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println(sql1); // 打印SQL语句
        tableEnv.executeSql(sql1);

        String sql2 = "CREATE TABLE behavior_gen (\n" +
                "    user_id STRING,\n" +
                "    item_id STRING,\n" +
                "    behavior BIGINT,\n" +
                "    s_diff AS UNIX_TIMESTAMP() - UNIX_TIMESTAMP('2025-01-27 14:14:00'),\n" +
                "    dt AS CASE \n" +
                "        WHEN (UNIX_TIMESTAMP() - UNIX_TIMESTAMP('2025-01-27 14:14:00')) > 0 \n" +
                "        THEN '2025-01-27'\n" +
                "        ELSE '2025-01-26' \n" +
                "    END,\n" +
                "    eventtime TIMESTAMP_LTZ(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen'\n" +
                "  , 'rows-per-second' = '100'\n" +
                ");";
        System.out.println(sql2); // 打印SQL语句
        tableEnv.executeSql(sql2);

        String dropTableSql = "DROP TABLE IF EXISTS `paimon`.`default`.`user_behavior`";
        System.out.println(dropTableSql); // 打印SQL语句
        tableEnv.executeSql(dropTableSql);

        String createTableSql = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_behavior` (\n" +
                "    user_id STRING,\n" +
                "    item_id STRING,\n" +
                "    behavior BIGINT,\n" +
                "    dt STRING,\n" +
                "    eventtime TIMESTAMP_LTZ(3),\n" +
                "    WATERMARK FOR eventtime AS eventtime - INTERVAL '5' SECOND,\n" +
                "    PRIMARY KEY (dt, user_id) NOT ENFORCED\n" +
                ") \n" +
                "PARTITIONED BY (dt)\n" +
                "WITH (\n" +
                "    'bucket' = '1',\n" +
                "    'changelog-producer' = 'none',\n" +
                "    'partition.expiration-time' = '99999 d',\n" +
                "    'partition.timestamp-formatter'='yyyy-MM-dd',\n" +
                "    'partition.timestamp-pattern'='$dt',\n" +
                "    'partition.time-interval'='1 d',\n" +
                "    'partition.idle-time-to-done'='1 m',\n" +
                "    'partition.mark-done-action'='success-file'\n" +
                ")\n" +
                ";";
        System.out.println(createTableSql); // 打印SQL语句
        tableEnv.executeSql(createTableSql);

//        String sql3 = "SELECT * \n" +
//                "FROM \n" +
//                "    behavior_gen\n" +
//                ";";
//        System.out.println(sql3);
//        tableEnv.executeSql(sql3).print();

        String insertSql = "INSERT INTO `paimon`.`default`.`user_behavior`\n" +
                "SELECT\n" +
                "    user_id\n" +
                "    , item_id\n" +
                "    , behavior\n" +
                "    , dt" +
                "    , eventtime\n" +
                "FROM \n" +
                "    behavior_gen\n" +
                ";";
        System.out.println(insertSql); // 打印SQL语句
        tableEnv.executeSql(insertSql);
    }
}
